package com.ctbri.manage.quality;
import com.ctbri.manage.quality.config.*;
import com.ctbri.manage.quality.config.YamlReader;
import com.ctbri.manage.quality.multi.DifferenceStatic2;
import com.ctbri.manage.quality.multi.TwoTablePerPosEqual;
import com.ctbri.manage.quality.single.*;
import com.ctbri.manage.quality.multi.*;
import org.apache.arrow.flatbuf.Bool;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import org.yaml.snakeyaml.Yaml;

/**
 运行所有稽查方法的主类
 @author xiaosonggege
 */
public class Main {
    /**
     * col null check 的excel版本
     * @param appName
     * @param masterSet
     * @param logLevel
     * @param dataSavePath
     * @param dataSaveStyle
     * @param needStructType
     * @param dataSource
     */
    private static void colNullCheck(final String appName,
                                          final String masterSet,
                                          final String logLevel,
                                          final String dataSavePath,
                                          final String dataSaveStyle,
                                          final Boolean needStructType,
                                          final String dataSource,
                                          final String dataExportPath,
                                          final String needCheckColName,
                                          final List<String> colName){
        CheckColumnNullType checkColumnNullType = new CheckColumnNullType(
                appName,
                masterSet,
                logLevel,
                dataExportPath,
                dataSavePath,
                dataSaveStyle
        );
        StructType structType = null;
        if (needStructType){
//            final List<String> colName = YamlReader.getInstance(configPath).getList("colName");
            List<StructField> structFields = new ArrayList<>();
            for (String i: colName) structFields.add(DataTypes.createStructField(i, DataTypes.StringType, true));
            structType = DataTypes.createStructType(structFields);
        }
        try{
            checkColumnNullType.call(structType, dataSource, needCheckColName);
        } catch (Exception e){
            e.printStackTrace();
        }
    }

    /**
     * col null check 的hive版本
     * @param appName
     * @param masterSet
     * @param logLevel
     * @param dataSavePath
     * @param dataSaveStyle
     * @param dataSource
     * @param databaseName
     * @param tableName
     * @param needCheckColName
     */
    private static void colNullCheck(final String appName,
                                          final String masterSet,
                                          final String logLevel,
                                          final String dataSavePath,
                                          final String dataSaveStyle,
                                          final String dataSource,
                                          final String databaseName,
                                          final String tableName,
                                          final String needCheckColName){
        CheckColumnNullType checkColumnNullType = new CheckColumnNullType(
                appName,
                masterSet,
                logLevel,
                databaseName,
                tableName,
                dataSavePath,
                dataSaveStyle
        );
        try{
            checkColumnNullType.call(null, dataSource, needCheckColName);
        } catch (Exception e){
            e.printStackTrace();
        }
    }

    /**
     * col str len check 的excel版本
     * @param appName
     * @param masterSet
     * @param logLevel
     * @param dataSavePath
     * @param dataSaveStyle
     * @param needStructType
     * @param dataSource
     * @param dataExportPath
     * @param needCheckColName
     * @param len
     * @param colName
     */
    private static void colStrLenCheck(final String appName,
                                      final String masterSet,
                                      final String logLevel,
                                      final String dataSavePath,
                                      final String dataSaveStyle,
                                      final Boolean needStructType,
                                      final String dataSource,
                                      final String dataExportPath,
                                      final String needCheckColName,
                                      final int len,
                                      final List<String> colName,
                                       final Boolean meetRequirement){
        CheckColumnStringLength checkColumnStringLength = new CheckColumnStringLength(
                appName,
                masterSet,
                logLevel,
                dataExportPath,
                dataSavePath,
                dataSaveStyle,
                len
        );
        StructType structType = null;
        if (needStructType){
            List<StructField> structFields = new ArrayList<>();
            for (String i: colName) structFields.add(DataTypes.createStructField(i, DataTypes.StringType, true));
            structType = DataTypes.createStructType(structFields);
        }
        try{
            checkColumnStringLength.call(structType, dataSource, needCheckColName, meetRequirement);
        } catch (Exception e){
            e.printStackTrace();
        }
    }

    /**
     * col str len check 的hive版本
     * @param appName
     * @param masterSet
     * @param logLevel
     * @param dataSavePath
     * @param dataSaveStyle
     * @param dataSource
     * @param databaseName
     * @param tableName
     * @param needCheckColName
     * @param len
     */
    private static void colStrLenCheck(final String appName,
                                      final String masterSet,
                                      final String logLevel,
                                      final String dataSavePath,
                                      final String dataSaveStyle,
                                      final String dataSource,
                                      final String databaseName,
                                      final String tableName,
                                      final String needCheckColName,
                                      final int len,
                                       final Boolean meetRequirement){
        CheckColumnStringLength checkColumnStringLength = new CheckColumnStringLength(
                appName,
                masterSet,
                logLevel,
                databaseName,
                tableName,
                len,
                dataSavePath,
                dataSaveStyle
        );
        try{
            checkColumnStringLength.call(null, dataSource, needCheckColName, meetRequirement);
        } catch (Exception e){
            e.printStackTrace();
        }
    }

    /**
     * col type check 的excel版本
     * @param appName
     * @param masterSet
     * @param logLevel
     * @param dataSavePath
     * @param dataSaveStyle
     * @param needStructType
     * @param dataSource
     * @param dataExportPath
     * @param needCheckColName
     * @param colName
     */
    private static void colTypeCheck(final String appName,
                                      final String masterSet,
                                      final String logLevel,
                                      final String dataSavePath,
                                      final String dataSaveStyle,
                                      final Boolean needStructType,
                                      final String dataSource,
                                      final String dataExportPath,
                                      final String needCheckColName,
                                      final List<String> colName,
                                      final Boolean meetRequirement){
        CheckColumnType checkColumnType = new CheckColumnType(
                appName,
                masterSet,
                logLevel,
                dataExportPath,
                dataSavePath,
                dataSaveStyle
        );
        StructType structType = null;
        if (needStructType){
            List<StructField> structFields = new ArrayList<>();
            for (String i: colName) structFields.add(DataTypes.createStructField(i, DataTypes.StringType, true));
            structType = DataTypes.createStructType(structFields);
        }
        try{
            checkColumnType.call(structType, dataSource, needCheckColName, meetRequirement);
        } catch (Exception e){
            e.printStackTrace();
        }
    }

    /**
     * col type check 的hive版本
     * @param appName
     * @param masterSet
     * @param logLevel
     * @param dataSavePath
     * @param dataSaveStyle
     * @param dataSource
     * @param databaseName
     * @param tableName
     * @param needCheckColName
     */
    private static void colTypeCheck(final String appName,
                                     final String masterSet,
                                     final String logLevel,
                                     final String dataSavePath,
                                     final String dataSaveStyle,
                                     final String dataSource,
                                     final String databaseName,
                                     final String tableName,
                                     final String needCheckColName,
                                     final Boolean meetRequirement){
        CheckColumnType checkColumnType = new CheckColumnType(
                appName,
                masterSet,
                logLevel,
                databaseName,
                tableName,
                dataSavePath,
                dataSaveStyle
        );

        try{
            checkColumnType.call(null, dataSource, needCheckColName, meetRequirement);
        } catch (Exception e){
            e.printStackTrace();
        }
    }

    /**
     * two tables diff check 的excel版本
     * @param appName
     * @param masterSet
     * @param logLevel
     * @param dataSavePath
     * @param dataSaveStyle
     * @param dataExportPath
     * @param dataExportPath2
     * @param joinColName1
     * @param joinColName2
     * @param joinMode
     * @param needStructType
     * @param dataSource
     */
    private static void twoTablesDiffCheck(final String appName,
                                           final String masterSet,
                                           final String logLevel,
                                           final String dataSavePath,
                                           final String dataSaveStyle,
                                           final String dataExportPath,
                                           final String dataExportPath2,
                                           final ArrayList<String> joinColName1,
                                           final ArrayList<String> joinColName2,
                                           final String joinMode,
                                           final Boolean needStructType,
                                           final String dataSource,
                                           final List<String> colName1,
                                           final List<String> colName2
                                           ){
        DifferenceStatic2 differenceStatic2 = new DifferenceStatic2(
                appName,
                masterSet,
                logLevel,
                dataExportPath,
                dataSavePath,
                dataSaveStyle,
                dataExportPath2,
                joinColName1,
                joinColName2,
                joinMode
        );
        StructType structType1 = null;
        StructType structType2 = null;
        if (needStructType){
            List<StructField> structFields1 = new ArrayList<>();
            for (String i: colName1) structFields1.add(DataTypes.createStructField(i, DataTypes.StringType, true));
            structType1 = DataTypes.createStructType(structFields1);

            List<StructField> structFields2 = new ArrayList<>();
            for (String i: colName2) structFields2.add(DataTypes.createStructField(i, DataTypes.StringType, true));
            structType2 = DataTypes.createStructType(structFields2);
        }
        try{
            differenceStatic2.call(structType1, structType2, dataSource, true, true);
        } catch (Exception e){
            e.printStackTrace();
        }
    }

    /**
     * two tables diff check 的hive版本
     * @param appName
     * @param masterSet
     * @param logLevel
     * @param dataSavePath
     * @param dataSaveStyle
     * @param databaseName
     * @param tableName1
     * @param tableName2
     * @param joinColName1
     * @param joinColName2
     * @param joinMode
     * @param dataSource
     */
    private static void twoTablesDiffCheck(final String appName,
                                           final String masterSet,
                                           final String logLevel,
                                           final String dataSavePath,
                                           final String dataSaveStyle,
                                           final String databaseName,
                                           final String tableName1,
                                           final String tableName2,
                                           final ArrayList<String> joinColName1,
                                           final ArrayList<String> joinColName2,
                                           final String joinMode,
                                           final String dataSource){
        DifferenceStatic2 differenceStatic2 = new DifferenceStatic2(
                appName,
                masterSet,
                logLevel,
                databaseName,
                tableName1,
                tableName2,
                joinColName1,
                joinColName2,
                joinMode,
                dataSavePath,
                dataSaveStyle
        );
        try{
            differenceStatic2.call(null, null, dataSource, true, true);
        } catch (Exception e){
            e.printStackTrace();
        }
    }

    /**
     * two tables per pos check 的excel版本
     * @param appName
     * @param masterSet
     * @param logLevel
     * @param dataSavePath
     * @param dataSaveStyle
     * @param dataExportPath
     * @param dataExportPath2
     * @param joinColName1
     * @param joinColName2
     * @param joinMode
     * @param needStructType
     * @param dataSource
     * @param needCheckCol
     */
    private static void twoTablesPerPosCheck(final String appName,
                                           final String masterSet,
                                           final String logLevel,
                                           final String dataSavePath,
                                           final String dataSaveStyle,
                                           final String dataExportPath,
                                           final String dataExportPath2,
                                           final ArrayList<String> joinColName1,
                                           final ArrayList<String> joinColName2,
                                           final String joinMode,
                                           final Boolean needStructType,
                                           final String dataSource,
                                           final List<ArrayList<String>> needCheckCol,
                                             final List<String> colName1,
                                             final List<String> colName2,
                                             final Boolean meetRequirements){
        TwoTablePerPosEqual ttppe = new TwoTablePerPosEqual(
                appName,
                masterSet,
                logLevel,
                dataExportPath,
                dataSavePath,
                dataSaveStyle,
                dataExportPath2,
                joinColName1,
                joinColName2,
                joinMode,
                needCheckCol
        );
        StructType structType1 = null;
        StructType structType2 = null;
        if (needStructType){
//            final List<String> colName1 = YamlReader.getInstance(configPath).getList("colName1");
            List<StructField> structFields1 = new ArrayList<>();
            for (String i: colName1) structFields1.add(DataTypes.createStructField(i, DataTypes.StringType, true));
            structType1 = DataTypes.createStructType(structFields1);

//            final List<String> colName2 = YamlReader.getInstance(configPath).getList("colName2");
            List<StructField> structFields2 = new ArrayList<>();
            for (String i: colName2) structFields2.add(DataTypes.createStructField(i, DataTypes.StringType, true));
            structType2 = DataTypes.createStructType(structFields2);
        }
        try {
            ttppe.call(structType1, structType2, dataSource, false, null, meetRequirements);
        } catch (Exception e){
            e.printStackTrace();
        }
    }

    /**
     * two tables per pos check 的hive版本
     * @param appName
     * @param masterSet
     * @param logLevel
     * @param dataSavePath
     * @param dataSaveStyle
     * @param databaseName
     * @param tableName1
     * @param tableName2
     * @param joinColName1
     * @param joinColName2
     * @param joinMode
     * @param dataSource
     * @param needCheckCol
     */
    private static void twoTablesPerPosCheck(final String appName,
                                             final String masterSet,
                                             final String logLevel,
                                             final String dataSavePath,
                                             final String dataSaveStyle,
                                             final String databaseName,
                                             final String tableName1,
                                             final String tableName2,
                                             final ArrayList<String> joinColName1,
                                             final ArrayList<String> joinColName2,
                                             final String joinMode,
                                             final String dataSource,
                                             final List<ArrayList<String>> needCheckCol,
                                             final Boolean meetRequirements){
        TwoTablePerPosEqual ttppe = new TwoTablePerPosEqual(
                appName,
                masterSet,
                logLevel,
                databaseName,
                tableName1,
                tableName2,
                joinColName1,
                joinColName2,
                joinMode,
                needCheckCol,
                dataSavePath,
                dataSaveStyle
        );

        try {
            ttppe.call(null, null, dataSource, false, null, meetRequirements);
        } catch (Exception e){
            e.printStackTrace();
        }
    }

    public static void main(String[] args) throws FileNotFoundException {
        //导入主配置文件信息
        final String filePathOfMainConfig = args[0];
        Map<String, Object> confMain = YamlReader.getInstance(filePathOfMainConfig).getConf();
        System.out.println("file path of main config:");
        System.out.println(confMain);
        final String checkName = YamlReader.getInstance(filePathOfMainConfig).getString("checkName");
        final String configPath = YamlReader.getInstance(filePathOfMainConfig).getString("configPath");
        final String dataSource1 = YamlReader.getInstance(filePathOfMainConfig).getString("dataSource");

        System.out.println("checkName: " + checkName);
        System.out.println("configPath: " + configPath);
        System.out.println("dataSource1: " + dataSource1);
       // YamlReader.getInstance(configPath).setConf((Map<String, Object>) new Yaml().load(new FileInputStream(new File(configPath))));
        //导入yaml配置信息
        Map<String, Object> conf = YamlReader.getInstance(configPath).getConf();
        System.out.println("file path of sub config:");
        System.out.println(conf);
        final String appName = YamlReader.getInstance(configPath).getString("appName");
        final String masterSet = YamlReader.getInstance(configPath).getString("masterSet");
        final String logLevel = YamlReader.getInstance(configPath).getString("logLevel");
        final String dataSavePath = YamlReader.getInstance(configPath).getString("dataSavePath");
        final String dataSaveStyle = YamlReader.getInstance(configPath).getString("dataSaveStyle");
//        final Boolean needStructType = YamlReader.getInstance(configPath).getBoolean("needStructType", false);
        final String dataSource = YamlReader.getInstance(configPath).getString("dataSource");
        final Boolean meetRequirement = YamlReader.getInstance(configPath).getBoolean("meetRequirement", false);
        //根据主配置文件信息进行数据源配置和数据需求分析
        if (checkName.equals("col_null_check")) {
            final String needCheckColName = YamlReader.getInstance(configPath).getString("needCheckColName");
            if (dataSource1.equals("excel")) {
                final String dataExportPath = YamlReader.getInstance(configPath).getString("dataExportPath");
                final Boolean needStructType = YamlReader.getInstance(configPath).getBoolean("needStructType", false);
                final List<String> colName = YamlReader.getInstance(configPath).getList("colName");
                Main.colNullCheck(appName, masterSet, logLevel, dataSavePath, dataSaveStyle, needStructType, dataSource, dataExportPath, needCheckColName, colName);
            }
            else if (dataSource1.equals("hive")) {
                final String databaseName = YamlReader.getInstance(configPath).getString("databaseName");
                final String tableName = YamlReader.getInstance(configPath).getString("tableName");
                Main.colNullCheck(appName, masterSet, logLevel, dataSavePath, dataSaveStyle, dataSource, databaseName, tableName, needCheckColName);
            }
           /* else if (false) {
                //todo: 其他数据源待实现
            }*/
        }
        else if (checkName.equals("col_str_len_check")) {
            final String needCheckColName = YamlReader.getInstance(configPath).getString("needCheckColName");
            if (dataSource1.equals("excel")) {
                final String dataExportPath = YamlReader.getInstance(configPath).getString("dataExportPath");
                final int len = YamlReader.getInstance(configPath).getInteger("len", false);
                final Boolean needStructType = YamlReader.getInstance(configPath).getBoolean("needStructType", false);
                final List<String> colName = YamlReader.getInstance(configPath).getList("colName");
                Main.colStrLenCheck(appName, masterSet, logLevel, dataSavePath, dataSaveStyle, needStructType, dataSource, dataExportPath, needCheckColName, len, colName, meetRequirement);
            }
            else if (dataSource1.equals("hive")){
                final String databaseName = YamlReader.getInstance(configPath).getString("databaseName");
                final String tableName = YamlReader.getInstance(configPath).getString("tableName");
                final int len = YamlReader.getInstance(configPath).getInteger("len", false);
                Main.colStrLenCheck(appName, masterSet, logLevel, dataSavePath, dataSaveStyle, dataSource, databaseName, tableName, needCheckColName, len, meetRequirement);
            }
          /*  else if (false) {
                //todo: 其他数据源待实现
            }*/
        }
        else if (checkName.equals("col_type_check")) {

            final String needCheckColName = YamlReader.getInstance(configPath).getString("needCheckColName");
            if (dataSource1.equals("excel")) {
                final String dataExportPath = YamlReader.getInstance(configPath).getString("dataExportPath");
                final Boolean needStructType = YamlReader.getInstance(configPath).getBoolean("needStructType", false);
                final List<String> colName = YamlReader.getInstance(configPath).getList("colName");
                Main.colTypeCheck(appName, masterSet, logLevel, dataSavePath, dataSaveStyle, needStructType, dataSource, dataExportPath, needCheckColName, colName, meetRequirement);
            } else if (dataSource1.equals("hive")) {
                final String databaseName = YamlReader.getInstance(configPath).getString("databaseName");
                final String tableName = YamlReader.getInstance(configPath).getString("tableName");
                Main.colTypeCheck(appName, masterSet, logLevel, dataSavePath, dataSaveStyle, dataSource, databaseName, tableName, needCheckColName, meetRequirement);
            }
           /* else if (false) {
                //todo: 其他数据源待实现
            }*/
        }
        else if (checkName.equals("two_tables_diff_check")) {
            final ArrayList<String> joinColName1 = YamlReader.getInstance(configPath).getList("joinColName1");
            final ArrayList<String> joinColName2 = YamlReader.getInstance(configPath).getList("joinColName2");
            final String joinMode = YamlReader.getInstance(configPath).getString("joinMode");
            if (dataSource1.equals("excel")) {
                final String dataExportPath = YamlReader.getInstance(configPath).getString("dataExportPath");
                final String dataExportPath2 = YamlReader.getInstance(configPath).getString("dataExportPath2");
                final Boolean needStructType = YamlReader.getInstance(configPath).getBoolean("needStructType", false);
                final List<String> colName1 = YamlReader.getInstance(configPath).getList("colName1");
                final List<String> colName2 = YamlReader.getInstance(configPath).getList("colName2");
                Main.twoTablesDiffCheck(appName, masterSet, logLevel, dataSavePath, dataSaveStyle, dataExportPath, dataExportPath2, joinColName1, joinColName2, joinMode, needStructType, dataSource, colName1, colName2);

            }
            else if (dataSource1.equals("hive")){
                final String databaseName = YamlReader.getInstance(configPath).getString("databaseName");
                final String tableName1 = YamlReader.getInstance(configPath).getString("tableName1");
                final String tableName2 = YamlReader.getInstance(configPath).getString("tableName2");
                Main.twoTablesDiffCheck(appName, masterSet, logLevel, dataSavePath, dataSaveStyle, databaseName, tableName1, tableName2, joinColName1, joinColName2, joinMode, dataSource);

            }
           /* else if (false) {
                //todo: 其他数据源待实现
            }*/
        }
        else if (checkName.equals("two_tables_per_pos_check")) {
            final ArrayList<String> joinColName1 = YamlReader.getInstance(configPath).getList("joinColName1");
            final ArrayList<String> joinColName2 = YamlReader.getInstance(configPath).getList("joinColName2");
            final String joinMode = YamlReader.getInstance(configPath).getString("joinMode");
            final List<ArrayList<String>> needCheckCol = YamlReader.getInstance(configPath).get2dimList("needCheckCol");
            if (dataSource1.equals("excel")) {
                final String dataExportPath = YamlReader.getInstance(configPath).getString("dataExportPath");
                final String dataExportPath2 = YamlReader.getInstance(configPath).getString("dataExportPath2");
                final Boolean needStructType = YamlReader.getInstance(configPath).getBoolean("needStructType", false);
                final List<String> colName1 = YamlReader.getInstance(configPath).getList("colName1");
                final List<String> colName2 = YamlReader.getInstance(configPath).getList("colName2");
                Main.twoTablesPerPosCheck(appName, masterSet, logLevel, dataSavePath, dataSaveStyle, dataExportPath, dataExportPath2, joinColName1, joinColName2, joinMode, needStructType, dataSource, needCheckCol, colName1, colName2, meetRequirement);
            }
            else if (dataSource1.equals("hive")){
                final String databaseName = YamlReader.getInstance(configPath).getString("databaseName");
                final String tableName1 = YamlReader.getInstance(configPath).getString("tableName1");
                final String tableName2 = YamlReader.getInstance(configPath).getString("tableName2");
                Main.twoTablesPerPosCheck(appName, masterSet, logLevel, dataSavePath, dataSaveStyle, databaseName, tableName1, tableName2, joinColName1, joinColName2, joinMode, dataSource, needCheckCol, meetRequirement);
            }
           /* else if (false) {
                //todo: 其他数据源待实现
            }*/
        }
       /* else if (false){
            //todo: 补充其他需求后填写
        }*/
        /*else {
            System.out.println("there is no specific function");
        }*/
    }
}
